package springboot.kafka.controller;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;
import java.util.Random;

//@Configuration
public class KafkaConsumerDemo {

//    @KafkaListener(topics = "cm",containerFactory = "batchFactory")
//    public void consumerTopic(List<ConsumerRecord> records){
//        System.out.println(records.size());
//        System.out.println("收到消息：" + records.toString());
//    }

    @KafkaListener(topics = "tsa-syslog")
    public void consumerTopic(List<ConsumerRecord> records, Acknowledgment ack) {
        System.out.println("收到的数据量个数：" + records.size());
        for (ConsumerRecord consumerRecord : records) {
            System.out.println("分区" + consumerRecord.partition() + "偏移量：" + consumerRecord.offset() + "属性值：" + consumerRecord.value());
        }

        Random random = new Random(2000);
        try {
            Thread.sleep(random.nextLong());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        ack.acknowledge();
    }
}
